import json
import os
import ssl

from kafka import KafkaProducer
from kafka.errors import KafkaError
from dialogue.service.dialogue_storage import dialogueStorage
import credit_voice_service.settings as settings
from credit_voice_service.settings import log

conf = settings.kafka_setting


class MyKafkaProducer:

    def __init__(self):
        context = ssl.create_default_context()
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        context.verify_mode = ssl.CERT_REQUIRED
        context.check_hostname = False
        context.load_verify_locations(os.getcwd() + "/only-4096-ca-cert")
        # 初始化一个Producer对象。
        self.producer = KafkaProducer(bootstrap_servers=conf['bootstrap_servers'],
                                      sasl_mechanism="PLAIN",
                                      ssl_context=context,
                                      security_protocol='SASL_SSL',
                                      api_version=(0, 10),
                                      acks=1,
                                      linger_ms=100,
                                      retries=5,
                                      sasl_plain_username=conf['sasl_plain_username'],
                                      sasl_plain_password=conf['sasl_plain_password'])

    def sendMessages(self, session_id, message):
        try:
            message_bytes = message.encode('utf-8')
            # 异步发送消息。
            self.producer.send(conf['topic_name'], message_bytes).add_errback(self.on_send_error, message)
            # 在程序结束时，调用flush。
            self.producer.flush()
        except Exception as e:
            dialogueStorage.mq_error_data(msg=message)
            log.info("[sendMessages] messages:{} session_id:{} error:{}".format(message, session_id, str(e)))

    def on_send_error(self, exception, message):
        log.info("[sendMessages] messages:{} 报错数据".format(message))
        dialogueStorage.mq_error_data(msg=message)

    if __name__ == '__main__':
        pass
